﻿using Misakai.Kafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace Reflexive.IO.Kafka
{
    public class KafkaTcpSocket : IKafkaTcpSocket, IDisposable
    {
        public delegate void ReconnectionAttempDelegate(int attempt);
        private const int DefaultReconnectionTimeout = 500;
        private const int DefaultReconnectionTimeoutMultiplier = 2;
        private readonly SemaphoreSlim _singleReaderSemaphore = new SemaphoreSlim(1, 1);
        private readonly ThreadWall _getConnectedClientThreadWall = new ThreadWall(ThreadWallState.Blocked);
        private readonly CancellationTokenSource _disposeToken = new CancellationTokenSource();
        private readonly IKafkaLog _log;
        private readonly KafkaEndpoint _endpoint;
        private TcpClient _client;
        private int _ensureOneThread;
        private int _disposeCount;
        private Task _clientConnectingTask;
        public event KafkaTcpSocket.ReconnectionAttempDelegate OnReconnectionAttempt;
        public KafkaEndpoint Endpoint
        {
            get
            {
                return this._endpoint;
            }
        }
        public KafkaTcpSocket(IKafkaLog log, KafkaEndpoint endpoint, int delayConnectAttemptMS = 0)
        {
            this._log = log;
            this._endpoint = endpoint;
            Task.Delay(TimeSpan.FromMilliseconds((double)delayConnectAttemptMS)).ContinueWith(delegate(Task x)
            {
                this.TriggerReconnection();
            });
        }
        public Task<byte[]> ReadAsync(int readSize)
        {
            return this.EnsureReadAsync(readSize, this._disposeToken.Token);
        }
        public Task<byte[]> ReadAsync(int readSize, CancellationToken cancellationToken)
        {
            return this.EnsureReadAsync(readSize, cancellationToken);
        }
        public Task WriteAsync(BinaryStream buffer)
        {
            return this.EnsureWriteAsync(buffer, this._disposeToken.Token);
        }
        public Task WriteAsync(BinaryStream buffer, CancellationToken cancellationToken)
        {
            return this.EnsureWriteAsync(buffer, cancellationToken);
        }
        private Task<TcpClient> GetClientAsync()
        {
            return this._getConnectedClientThreadWall.RequestPassageAsync().ContinueWith<TcpClient>(delegate(Task t)
            {
                if (this._client == null)
                {
                    throw new ServerUnreachableException(string.Format("Connection to {0} was not established.", this._endpoint), new object[0]);
                }
                return this._client;
            });
        }
        private async Task EnsureWriteAsync(BinaryStream buffer, CancellationToken cancellationToken)
        {
            try
            {
                TcpClient tcpClient = await this.GetClientAsync();
                int count = await buffer.Flush();
                await tcpClient.GetStream().WriteAsync(buffer.GetBuffer(), 0, count, cancellationToken);
            }
            catch
            {
                if (this._disposeToken.IsCancellationRequested)
                {
                    throw new ObjectDisposedException("Object is disposing.");
                }
                throw;
            }
        }
        private async Task<byte[]> EnsureReadAsync(int readSize, CancellationToken token)
        {
            CancellationTokenRegistration cancellationTokenRegistration = default(CancellationTokenRegistration);
            byte[] result;
            try
            {
                await this._singleReaderSemaphore.WaitAsync(token);
                byte[] array = new byte[readSize];
                int i = 0;
                int num = 0;
                while (i < readSize)
                {
                    readSize -= i;
                    TcpClient tcpClient = await this.GetClientAsync();
                    i = await tcpClient.GetStream().ReadAsync(array, num, readSize, token).WithCancellation(token);
                    if (i <= 0)
                    {
                        this.TriggerReconnection();
                        throw new ServerDisconnectedException(string.Format("Lost connection to server: {0}", this._endpoint), new object[0]);
                    }
                    num += i;
                }
                result = array;
                
            }
            catch(System.InvalidOperationException)
            {
                if (this._disposeToken.IsCancellationRequested)
                {
                    throw new ObjectDisposedException("Object is disposing.");
                }
                this.TriggerReconnection();
                throw;
            }
            catch (System.IO.IOException)
            {
                if (this._disposeToken.IsCancellationRequested)
                {
                    throw new ObjectDisposedException("Object is disposing.");
                }
                this.TriggerReconnection();
                throw;
            }
            catch
            {
                if (this._disposeToken.IsCancellationRequested)
                {
                    throw new ObjectDisposedException("Object is disposing.");
                }
                throw;
            }
            finally
            {
                cancellationTokenRegistration.Dispose();
                this._singleReaderSemaphore.Release();
            }
            return result;
        }
        private void TriggerReconnection()
        {
            if (Interlocked.Increment(ref this._ensureOneThread) == 1)
            {
                this._getConnectedClientThreadWall.Block();
                this._clientConnectingTask = this.ReEstablishConnectionAsync().ContinueWith(delegate(Task<TcpClient> x)
                {
                    Interlocked.Decrement(ref this._ensureOneThread);
                    this._getConnectedClientThreadWall.Release();
                });
                return;
            }
            Interlocked.Decrement(ref this._ensureOneThread);
        }
        private async Task<TcpClient> ReEstablishConnectionAsync()
        {
            int num = 1;
            int delay = DefaultReconnectionTimeout;
            this._log.WarnFormat("No connection to:{0}.  Attempting to re-connect...", new object[]
			{
				this._endpoint
			});

            TcpClient client2;
            using (this._client)
            {
                this._client = null;
                client2 = this._client;
            }

            while (!this._disposeToken.IsCancellationRequested)
            {
                try
                {
                    if (this.OnReconnectionAttempt != null)
                    {
                        KafkaTcpSocket.ReconnectionAttempDelegate reconnectionDelegate = this.OnReconnectionAttempt;
                        int attempt;
                        num = (attempt = num) + 1;
                        reconnectionDelegate(attempt);
                    }
                    this._client = new TcpClient();
                    await this._client.ConnectAsync(this._endpoint.Endpoint.Address, this._endpoint.Endpoint.Port);
                    this._log.WarnFormat("Connection established to:{0}.", new object[]
				    {
					    this._endpoint
				    });
                    client2 = this._client;
                    return client2;
                }
                catch
                {
                    delay *= DefaultReconnectionTimeoutMultiplier;
                    this._log.WarnFormat("Failed re-connection to:{0}.  Will retry in:{1}", new object[]
				    {
					    this._endpoint,
					    delay
				    });
                }
                await Task.Delay(TimeSpan.FromMilliseconds((double)delay), this._disposeToken.Token);
            }

            return client2;
        }
        public void Dispose()
        {
            if (Interlocked.Increment(ref this._disposeCount) != 1)
            {
                return;
            }
            if (this._disposeToken != null)
            {
                this._disposeToken.Cancel();
            }
            using (this._disposeToken)
            {
                using (this._client)
                {
                    if (this._clientConnectingTask != null)
                    {
                        this._clientConnectingTask.Wait(TimeSpan.FromSeconds(5.0));
                    }
                }
            }
        }
    }
}
